package com.cryptoneedle.garden.service.seatunnel;

import com.cryptoneedle.garden.entity.source.metadata.SourceCatalog;
import com.cryptoneedle.garden.entity.source.metadata.SourceDatabase;
import com.cryptoneedle.garden.entity.source.metadata.SourceTable;
import com.cryptoneedle.garden.service.crud.garden.source.*;
import com.cryptoneedle.garden.service.util.ConfigUtil;
import jakarta.annotation.Resource;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/**
 * <p>description:  </p>
 *
 * @author CryptoNeedle
 * @date 2025-11-05
 */
@Service
public class SeatunnelService {

    @Autowired
    private SeatunnelService seatunnelService;

    @Autowired
    private FoundSourceService select;
    @Autowired
    private AddSourceService add;
    @Autowired
    private PatchSourceService patch;
    @Autowired
    private SaveSourceService save;
    @Autowired
    private DeleteSourceService delete;

    @Resource
    private ConfigUtil configUtil;

    public String seatunnelScript(String catalogName, String databaseName, String tableName) {
        SourceCatalog catalog = select.catalog(catalogName);
        SourceDatabase database = select.database(catalogName, databaseName);
        SourceTable table = select.table(catalogName, databaseName, tableName);

        return seatunnelService.seatunnelScript(catalog, database, table);
    }

    public String seatunnelScript(SourceCatalog catalog, SourceDatabase database, SourceTable table) {
        StringBuilder script = new StringBuilder();

        // env
        script.append("""
                env {
                  job.mode = "BATCH"
                  shade.identifier = "base64"
                }
                
                """);

        // Source
        script.append("""
                source {
                  Jdbc {
                    result_table_name = "source_data"
                    url = "%s"
                    driver = %s
                    user = "%s"
                    password = "%s"
                    connection_check_timeout_sec = 100000
                    fetch_size = 5000
                    query = "SELECT * FROM %s.%s"
                    #where_condition = "WHERE 1 = 1"
                  }
                }
                
                """.formatted(catalog.getUrl(),
                catalog.getDriverClass(),
                catalog.getUserName(),
                catalog.getPassword(),
                table.getDatabaseName(),
                table.getTableName()));

        // Transform
        script.append("""
                transform {
                  FieldRename {
                    plugin_input = "source_data"
                    convert_case = "LOWER"
                    plugin_output = "lower_field_transform_data"
                  }
                  SQL {
                    source_table_name = "lower_field_transform_data"
                    result_table_name = "out_data"
                    query = "SELECT * FROM lower_field_transform_data"
                    plugin_output = "transform_data"
                  }
                }
                
                """);

        // Sink
        script.append("""
                sink {
                   Doris {
                    source_table_name = "transform_data"
                    fenodes = "%s:%s"
                    username = "%s"
                    password = "%s"
                    database = %s
                    doris.batch.size = 20000
                    data_save_mode = "APPEND_DATA"
                    table = "%s"
                    sink.label-prefix = "%s"
                    sink.enable-2pc = "false"
                    doris.config {
                      format = "json"
                      read_json_by_line = "true"
                    }
                  }
                }
                """.formatted("",
                "",
                "",
                "",
                configUtil.ods(),
                table.getTransTableName(),
                table.getTransTableName()));

        return script.toString();
    }
}